/*
 * FcfThread.java
 *
 * Created on August 3, 2008, 10:12 PM
 *
 */

package orajdbcstat;

import java.io.*;
import java.net.*;
import java.util.*;
import java.sql.*;
import java.math.*;
import oracle.jdbc.driver.*;
import oracle.jdbc.pool.*;
import oracle.jdbc.rowset.*;

/**
 *
 * @author vnull
 */
public class FcfThread implements Runnable {
    
    Random randomGenerator = new Random();
    Integer statementInstance = new Integer(-1);
    Integer statementInstanceOld = new Integer(-1);
    Long transactionTime = new Long(0);
    Long execTime = new Long(0);
    
    Long trx_commits = new Long(0);
    Long trx_rollbacks = new Long(0);
    Long old_trx_commits = new Long(0);
    Long old_trx_rollbacks = new Long(0);
    
    Boolean ok = true;
    Instance instance = null;
    Configuration conf = null;
    
    OracleDataSource ods = null;
    Connection statementDb = null;
    
    /* Creates a new instance of FcfThread */
    public FcfThread(Instance inst, Configuration conf) {
        this.instance = inst;
        this.conf = conf;
        
        try {
            /* Prepare pool */
            Properties cacheProps = new Properties();
            
            /* Low watermark for physical connections to the database in pool */
            cacheProps.setProperty("MinLimit", "2");
            /* High watermark for physical connections to the database in pool */
            cacheProps.setProperty("MaxLimit", "5");
            /* Initial number of physical connections to the database in pool */
            cacheProps.setProperty("InitialLimit", "2");
            //cacheProps.setProperty("ConnectionWaitTimeout", conf.networkDelay.toString());
            //cacheProps.setProperty("ConnectionWaitTimeout", "1800");
            cacheProps.setProperty("ValidateConnection", "true");
            
            /*
             * This property enables the JDBC client to quickly traverse an address_list
             * in the event of a failure. For example, if the client attempts to connect
             * to a host that is unavailable, the connection attempt will be
             * bounded to the time specified by the SQLnetDef.TCP_CONNTIMEOUT_STR
             * property after which the client attempts to connect to the
             * next host in the address_list. The behavior continues for each host
             * in the address_list until a connection is made. Setting the property
             * to a value of 3 seconds will suffice in most environments. It is
             * important to note that the SQLnetDef.TCP_CONNTIMEOUT_STR property
             * should be set on the data source and not on the implicit connection
             * cache.
             */

            cacheProps.put(oracle.net.ns.SQLnetDef.TCP_CONNTIMEOUT_STR, new Integer(conf.networkDelay * 1000).toString());
            cacheProps.put("oracle.net.READ_TIMEOUT", conf.networkDelay.toString());
            

            ods = new OracleDataSource();
            ods.setConnectionCacheProperties(cacheProps);
            ods.setConnectionCacheName(this.instance.TNS);
            ods.setUser(conf.login);
            ods.setPassword(conf.pass);
            ods.setDriverType("oci8");
            ods.setTNSEntryName(this.instance.TNS);
            ods.setConnectionCachingEnabled(true);
                /*
                 * Sets the maximum time in seconds that this datasource will wait
                 * while attempting to connect to the database
                 */
            //ods.setLoginTimeout(conf.networkDelay);

            if(conf.verbose >= 12) {
                System.err.println("Initializing OracleDataSource for " + this.instance.TNS +
                        " TCP_CONNTIMEOUT=" + cacheProps.get(oracle.net.ns.SQLnetDef.TCP_CONNTIMEOUT_STR) +
                        " TCP_READTIMEOUT=" + cacheProps.get(oracle.net.ns.SQLnetDef.TCP_READTIMEOUT_STR) +
                        " ConnectionWaitTimeout=" + cacheProps.getProperty("ConnectionWaitTimeout") +
                        " ods.getLoginTimeout()=" + ods.getLoginTimeout());
            }
            
            if(conf.verbose >= 2) {
                System.err.println("FcfThread::init["+this.instance.TNS+"] connecting to TNS "+this.instance.TNS);
            }
            
            if(conf.ONSConfiguration != null) {
                ods.setONSConfiguration(conf.ONSConfiguration);
                ods.setFastConnectionFailoverEnabled(true);
            }
            
            /*
             * First call to getConnection() can take several seconds,
             * so we have initialize "ods" first.
             */
            statementDb = ods.getConnection();
            statementDb.setAutoCommit(false);
            
        } catch (SQLException e) {
            /* we can fail here ? */
            System.err.println("FcfThread::init["+instance.TNS+"] " + e.getMessage());
            /* ORA-12571 TNS packet writer failure, happens e.g. when we are getting ECONNREFUSED
             * because on of the RAC nodes is online, but TCP rejects connections to listener
             * (listener is not listening on this alias, typical under failure conditions 
             */
            if(e.getErrorCode() != 12571 && conf.verbose >= 10 ){
                e.printStackTrace();
                System.exit(2);
            }
        }
    }
    
    public Integer getStatementInstance() {
        return statementInstance;
    }
    
    public Integer getOldStatementInstance() {
        return statementInstanceOld;
    }
    
    
    public Long getTransactionTime() {
        return transactionTime;
    }
    
    public Long getExecTime() {
        return execTime;
    }
    
    public Long getCommits() {
        return trx_commits;
    }
    
    public Long getRollbacks() {
        return trx_rollbacks;
    }
    
    public void DDLStatement(Connection db, String text) throws SQLException {
        if(db != null) {
            if(conf.verbose >= 9) {
                System.err.println("[DDL]: " + text);
            }
            Statement s = db.createStatement();
            s.executeUpdate(text);
            s.close();
        } else {
            System.err.println("Database is not connected for DDL: " + text);
            if(conf.verbose >= 10) {
                System.exit(2);
            }
        }
    }
    
    public void recheckStatementDb() {
        if(statementDb == null) {
            try {
                statementDb = ods.getConnection();
                statementDb.setAutoCommit(false);
                
            } catch(SQLException e) {
                statementDb = null;
            }
        }
    }
    
    
    
    public void run() {
        Random myRand = new Random();
        
        transactionTime = execTime = new Long(0);
        
        while(ok) {
            /* whole "monitoring loop" measurement */
            Long timerStart = System.currentTimeMillis();
            try {
                
                recheckStatementDb();
                
                if (statementDb != null) {
                    Statement s = statementDb.createStatement();

                    if(conf.verbose >= 10) {
                        System.err.println("FcfThread::run[" + this.instance.TNS + "] starting work");
                    }


                    /* fetch Instance ID (for RAC failover purposes) */
                    ResultSet r = s.executeQuery("SELECT SYS_CONTEXT('USERENV', 'INSTANCE') FROM DUAL");
                    r.next();
                    statementInstanceOld = statementInstance;
                    statementInstance = r.getInt(1);
                    r.close();
                    
                    /* calculate commit and rollback ratios, we NEED access to V_$SYSTAT (GRANT) */
                    r = s.executeQuery("SELECT name, value FROM SYS.V_$SYSSTAT WHERE name IN ('user commits', 'user rollbacks')");
                    while(r.next()) {
                        String name = r.getString(1);
                        Long value = r.getLong(2);
                        if(name.equals("user commits")) {
                            trx_commits = new Long(value - old_trx_commits);
                            old_trx_commits = value;
                        } else if (name.equals("user rollbacks")) {
                            trx_rollbacks = new Long(value - old_trx_rollbacks);
                            old_trx_rollbacks = value;
                            
                        }
                    }
                    r.close();
                    
                    /* perform several iterations for correct measurements */
                    for(int i = 0; i < conf.iterations; i++) {
                        Long startTimestamp;
                        
                        /* our new id (primary key) & PLSQL proc argument */
                        Integer base = conf.minRand;
                        Integer id = base + randomGenerator.nextInt(conf.maxRand);
                        if(conf.verbose >= 9) {
                            System.err.println("FcfThread::run[" + this.instance.TNS + "] id=" +
                                    id+" iteration="+i);

                        }
                        
                        /* First DELETE this ID to have clean state, but don't measure it
                         * this way we might avoid ID conflict in next insert step
                         */
                        DDLStatement(statementDb, "DELETE FROM " + conf.table + " WHERE id='"+id+"'");
                        
                        String myRandTxt = Long.toString(Math.abs(myRand.nextLong()), 60);
                        String myRandTxtUpdate = Long.toString(Math.abs(myRand.nextLong()), 60);
                        
                        startTimestamp = System.currentTimeMillis();
                        
                        /* Then INSERT our fresh ID with random data */
                        DDLStatement(statementDb, "INSERT INTO " + conf.table + "(id, data1) VALUES ('"+ id +"', '" + myRandTxt + "')");
                        
                        /* UPDATE our just-inserted ID */
                        DDLStatement(statementDb, "UPDATE " + conf.table + " SET data1 = '" + myRandTxtUpdate + "' WHERE id='"+id+"'");
                        
                        /* SELECT our fresh id as last step*/
                        Statement selectS = statementDb.createStatement();
                        r = selectS.executeQuery("SELECT id, data1 FROM " + conf.table + " WHERE id='" + id + "'");
                        while(r.next()) {
                            String str = r.getString(2);
                        }
                        r.close();
                        
                        /* DELETE our record(id) */
                        DDLStatement(statementDb, "DELETE FROM " + conf.table + " WHERE id='"+id+"'");
                        
                        statementDb.commit();
                        
                        transactionTime = System.currentTimeMillis() - startTimestamp;
                        
                        /* PL/SQL */
                        if(conf.plsqlProc != null && !conf.plsqlProc.equals("")) {
                            if(conf.verbose >= 9) {
                                System.err.println("FcfThread::run id="+id+" iteration="+i+ " launching PLSQL procedure");
                            }
                            startTimestamp = System.currentTimeMillis();
                            /* DDLStatement(statementDb, "EXEC " + conf.plsqlProc + "('"+ id + "')"); */
                            CallableStatement estmt = statementDb.prepareCall("{CALL " + conf.plsqlProc + "(?)}");
                            estmt.setInt(1, id);
                            estmt.execute();
                            estmt.close();
                            
                            statementDb.commit();
                            
                            execTime = System.currentTimeMillis() - startTimestamp;
                        }
                        
                    }
                    
                    /* average timings */
                    transactionTime /= conf.iterations;
                    execTime /= conf.iterations;
                    
                    s.close();
                    
                    
                    /*
                    OracleConnectionCacheManager occm = OracleConnectionCacheManager.getConnectionCacheManagerInstance();
                    String cacheName = ods.getConnectionCacheName();
                    int act = occm.getNumberOfActiveConnections(cacheName);
                    int avail = occm.getNumberOfAvailableConnections(cacheName);
                    
                    System.out.println(act+"/"+avail);
                    */
                    
                } else {
                    statementInstanceOld = statementInstance;
                    statementInstance = -1;
                    //selectTime = deleteTime = updateTime = insertTime = new Long(0);
                }
                
            } catch (SQLException e) {
                transactionTime = new Long(-e.getErrorCode());
                
                if(conf.plsqlProc != null && !conf.plsqlProc.equals("")) {
                    execTime = new Long(-e.getErrorCode());
                }
                
                if(conf.verbose >= 1) {
                    System.err.println("STMTCONN/THREAD["+instance.TNS+"] " + e.getMessage());

                    if(conf.verbose >= 10) {
                        e.printStackTrace();
                        //System.exit(1);
                    }
                }
                
                try {
                    statementDb.close();
                    if(conf.verbose >= 3) {
                        System.err.println("STMTCONN/THREAD["+instance.TNS+"] disconnected");
                    }
                } catch(SQLException eClose) {
                }
                
                /* mark statementDb as invalid */
                statementDb = null;
                if(statementInstance == -1) {
                    statementInstanceOld = -1;
                } else {
                    statementInstance = -1;
                }
            }
            
            
            try {
                Long timeUsed = System.currentTimeMillis() - timerStart;
                Long timeToSleep = Math.max((conf.delay*1000) - timeUsed, 0);
                if(timeToSleep <= 0) {
                    conf.delay = Math.min(conf.maxDelay, timeUsed.intValue()/1000 + 1);
                    timeToSleep = new Long(0);
                    System.err.println("STMTCONN/THREAD["+instance.TNS+"] Main refresh thread running too fast, new sleep time is " +
                            conf.delay + "s");
                }
                if(conf.verbose >= 3) {
                    System.err.println("STMTCONN/THREAD["+instance.TNS+"] sleeping " + timeToSleep + "ms " +
                            "(real time used = " + timeUsed+ "ms)");
                }
                
                Thread.sleep(timeToSleep);
            } catch(InterruptedException e) {
            }
            
        }
    }
    
}
